/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2019-2022. All rights reserved.
 * Description:hcom执行器
 * Author:wangwenwen
 * Create:2019-11-25
 */

#include "hcom_executor_internel.h"
#include <thread>
#include "hccl/hcom.h"
#include "hcom_pub.h"
#include "comm.h"
#include "op_base/src/op_base.h"
#include "stream_pub.h"
#include "workflow_pub.h"
#include "op_base_pub.h"
#include "hccl_comm_pub.h"
#include "param_check_pub.h"
#include "op_hcom_comm.h"
#include <nlohmann/json.hpp>
#include "graph/ge_local_context.h"
#include "framework/common/ge_types.h" // ge对外options
#include "sal.h"
#include "mem_host_pub.h"
#include "op_base_rpc.h"

HcclResult HcomExecInitialize()
{
    HCCL_INFO("Hcom Excutor Initialize start.");
    HcclResult ret = hccl::HcomExecutor::GetInstance().Initialize();
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Initialize][HcomExecutor]errNo[0x%016llx]initialize hcom executor failed",
            HCCL_ERROR_CODE(ret)), ret);
    HCCL_RUN_INFO("[Initialize][HcomExecutor]Hcom Excutor Initialize end. ret[%d]", ret);
    return ret;
}

HcclResult HcomExecFinalize()
{
    HCCL_INFO("Hcom Excutor Finalize start.");
    HcclResult ret = hccl::HcomExecutor::GetInstance().Finalize();
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Finalize][HcomExecutor]errNo[0x%016llx]finalize hcom executor failed",
            HCCL_ERROR_CODE(ret)), ret);
    HCCL_RUN_INFO("[Finalize][HcomExecutor]Hcom Excutor Finalize end. ret[%d]", ret);
    return ret;
}

HcclResult HcomExecEnqueueOperation(struct HcomOperation opInfo, std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueueOperation:hcclType[%s], inputPtr[%p], outputPtr[%p], count[%llu], "\
        "dataType[%s], op[%s]", opInfo.hcclType.c_str(), opInfo.inputPtr, opInfo.outputPtr, opInfo.count,
        GetDataTypeEnumStr(opInfo.dataType).c_str(), GetReduceOpEnumStr(opInfo.opType).c_str());
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueOperation(opInfo, callback);
}

HcclResult HcomExecEnqueueRemoteOperation(struct HcomRemoteOperation opInfo,
                                          std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueuRemoteeOperation:opdesc[%p], keyAddr[%p], tableId[%p], value[%p]", \
        opInfo.opdesc, opInfo.keyAddr, opInfo.tableId, opInfo.value);
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueRemoteOperation(opInfo, callback);
}

HcclResult HcomExecEnqueueRemoteAccess(const std::string& remoteAccessType,
                                       const std::vector<HcomRemoteAccessAddrInfo>& addrInfos,
                                       std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueueRemoteAccess:RemoteAccessType[%s] AddrInfos Size[%u]",
        remoteAccessType.c_str(), addrInfos.size());
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueRemoteAccess(remoteAccessType, addrInfos, callback);
}

HcclResult HcomExecEnqueueAllToAllV(HcomAllToAllVParams params, std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueueAllToAllV:AlltoAllV");
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueAllToAllV(params, callback);
}

HcclResult HcomExecEnqueueAllToAllVC(HcomAllToAllVCParams params, std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueueAllToAllVC:AlltoAllVC");
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueAllToAllVC(params, callback);
}

HcclResult HcomExecEnqueueGatherAllToAllV(HcomGatherAllToAllVParams params,
                                          std::function<void(HcclResult status)> callback)
{
    HCCL_RUN_INFO("Entry-HcomExecEnqueueGatherAllToAllV:Gather AlltoAllV");
    return hccl::HcomExecutor::GetInstance().HcomExecEnqueueGatherAllToAllV(params, callback);
}

namespace hccl {
namespace {
    const MsgQueueType MSGQUE_TYPE[] = {MsgQueueType::OPBASE_QUEUE, MsgQueueType::REMOTE_ACCESS_QUEUE};
    const u32 GATHER_THREAD_NUM = 16;
    const u32 NUM_TWO = 2;
}

HcomExecutor::HcomExecutor()
    : deviceLogicId_(-1),
      deviceType_(DevType::DEV_TYPE_910),
      msgMutex_(),
      eventMutex_(),
      parralMap_(),
      executorInitedFlag_(ATOMIC_FLAG_INIT),
      remoteAccess_(nullptr)
{}

HcomExecutor::~HcomExecutor()
{
    CleanQueueResources();
}

HcclResult HcomExecutor::InitGroup()
{
    std::string groupConf;
    nlohmann::json groupListConf;
    ge::graphStatus geRet = ge::GetThreadLocalContext().GetOption(ge::OPTION_EXEC_HCOM_GROUPLIST, groupConf);
    CHK_PRT_RET((geRet != ge::GRAPH_SUCCESS),
        HCCL_WARNING("[HcomOpsKernelInfoStore][InitHcom]OPTION_EXEC_HCOM_GROUPLIST is not found."), HCCL_SUCCESS);

    HCCL_DEBUG("groupList:%s", groupConf.c_str());
    CHK_RET(SalParseInformation(groupListConf, groupConf));
    std::vector<nlohmann::json> groupList = groupListConf.get<std::vector<nlohmann::json>>();
    for (auto &groupInfo : groupList) {
        HCCL_DEBUG("groupInfo:%s", groupInfo.dump().c_str());
        std::vector<u32> ranks = groupInfo["group_rank_list"];
        std::string groupName = groupInfo["group_name"];
        HCCL_DEBUG("groupName:%s", groupName.c_str());

        u32 curRank = 0;
        CHK_RET(HcomGetRankId(HCCL_WORLD_GROUP, &curRank));
        if (!HcomFindGroup(groupName.c_str()) && find(ranks.begin(), ranks.end(), curRank) != ranks.end()) {
            if (!strncmp(groupName.c_str(), HCCL_WORLD_GROUP, sizeof(HCCL_WORLD_GROUP))) {
                HCCL_WARNING("[HcomOpsKernelInfoStore][InitHcom]cur groupname is HCCL_WORLD_GROUP.");
                continue;
            }
            CHK_RET(HcomCreateGroup(groupName.c_str(), ranks.size(), ranks.data()));
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::Initialize()
{
    HcclResult ret;
    CHK_RET(hrtGetDevice(&deviceLogicId_));
    CHK_RET(hrtGetDeviceType(deviceType_));

    CHK_RET(HcomInitialize());

    CHK_RET(InitGroup());

    // 执行器创建stream
    parralMap_[MsgQueueType::OPBASE_QUEUE].stream.reset(new (std::nothrow)
        Stream(StreamType::STREAM_TYPE_ONLINE));
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::OPBASE_QUEUE].stream);

    parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream.reset(new (std::nothrow)
        Stream(StreamType::STREAM_TYPE_ONLINE));
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream);

    bool errorFlag = false;
    do {
        workflowMode_ = GetWorkflowMode();
        // 启动消息处理线
        parralMap_[MsgQueueType::OPBASE_QUEUE].messageProThread.reset(
            new (std::nothrow) std::thread(&HcomExecutor::MessageProcessThreadLoop, this,
                MsgQueueType::OPBASE_QUEUE));
        CHK_PRT_BREAK(!parralMap_[MsgQueueType::OPBASE_QUEUE].messageProThread,
            HCCL_ERROR("[Initialize][HcomExecutor]OPBASED MessageProcess threadPtr is Null"), errorFlag = true);

        parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].messageProThread.reset(
            new (std::nothrow) std::thread(&HcomExecutor::MessageProcessThreadLoop, this,
                MsgQueueType::REMOTE_ACCESS_QUEUE));
        CHK_PRT_BREAK(!parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].messageProThread,
            HCCL_ERROR("[Initialize][HcomExecutor]REMOTE_ACCESS MessageProcess threadPtr is Null"), errorFlag = true);

        // 启动回调处理线程
        parralMap_[MsgQueueType::OPBASE_QUEUE].callbackNotifyThread.reset(
            new (std::nothrow) std::thread(&HcomExecutor::CallbackNotifyThreadLoop, this,
                MsgQueueType::OPBASE_QUEUE));
        CHK_PRT_BREAK(!parralMap_[MsgQueueType::OPBASE_QUEUE].callbackNotifyThread,
            HCCL_ERROR("[Initialize][HcomExecutor]OPBASED callback Notify threadPtr is Null"), errorFlag = true);

        parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].callbackNotifyThread.reset(
            new (std::nothrow) std::thread(&HcomExecutor::CallbackNotifyThreadLoop, this,
                MsgQueueType::REMOTE_ACCESS_QUEUE));
        CHK_PRT_BREAK(!parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].callbackNotifyThread,
            HCCL_ERROR("[Initialize][HcomExecutor]REMOTE_ACCESS callback Notify threadPtr is Null"), errorFlag = true);

        /* 初始化竞争, 只允许被初始化一次 */
        ret = AtomicInitSet();
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Initialize][HcomExecutor]errNo[0x%016llx] repeat initialized",
            HCCL_ERROR_CODE(ret)), errorFlag = true);
    } while (0);

    if (errorFlag) {
        HCCL_ERROR("[Initialize][HcomExecutor]hcom executor init failed");
        if (Finalize() != HCCL_SUCCESS) {
            HCCL_ERROR("[Initialize][HcomExecutor]hcom finalize failed");
        }
        return HCCL_E_INTERNAL;
    }

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::InitHcclComm(const char *group, HcclComm comm)
{
    CHK_PRT_RET((comm == nullptr), HCCL_ERROR("comm is not initialized"), HCCL_E_PTR);
    if (comms_.find(comm) == comms_.end()) {
        // 申请device内存
        hccl::hcclComm* hcclComm = static_cast<hccl::hcclComm *>(comm);
        void *inCCLBufferPtr = nullptr;
        u64 inCCLBufferSize = 0;
        void *outCCLBufferPtr = nullptr;
        u64 outCCLBufferSize = 0;
        CHK_RET(hcclComm->GetInCCLbuffer(inCCLBufferPtr, inCCLBufferSize));
        CHK_RET(hcclComm->GetOutCCLbuffer(outCCLBufferPtr, outCCLBufferSize));
        if (inCCLBufferPtr == nullptr || outCCLBufferPtr == nullptr) {
            CHK_RET(hcclComm->CreateCommCCLbuffer());
        }
        comms_.insert(comm);
    }
    return HCCL_SUCCESS;
}

HcomExecutor &HcomExecutor::GetInstance()
{
    static HcomExecutor executor;
    return executor;
}

HcclResult HcomExecutor::AtomicInitSet()
{
    CHK_PRT_RET(executorInitedFlag_.test_and_set(),
        HCCL_ERROR("[AtomicInitSet][HcomExecutor]errNo[0x%016llx] executor already been initialized", \
            HCCL_ERROR_CODE(HCCL_E_INTERNAL)), HCCL_E_INTERNAL);
    return HCCL_SUCCESS;
}

void HcomExecutor::AtomicInitClear()
{
    executorInitedFlag_.clear();
}

HcclResult HcomExecutor::Finalize()
{
    // 清理初始化标志位
    AtomicInitClear();
    for (auto &key : MSGQUE_TYPE) {
        parralMap_[key].shutDown = true;
        if (parralMap_[key].messageProThread != nullptr) {
            parralMap_[key].messageProThread->join();
        }
        if (parralMap_[key].callbackNotifyThread != nullptr) {
            parralMap_[key].callbackNotifyThread->join();
        }
    }
    CleanQueueResources();

    // 销毁remoteaccess资源
    if (remoteAccess_ != nullptr) {
        remoteAccess_.reset();
    }
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueOperation(HcomOperation_t opInfo, StatusCallback callback)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(InitHcclComm(opInfo.group, comm));

    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::OPBASE_QUEUE].stream);

    if (parralMap_[MsgQueueType::OPBASE_QUEUE].shutDown) {
        HCCL_ERROR("[HcomExec][EnqueueOperation]thread has shut down, Enqueue hcom operation fail, op type[%s]",\
            opInfo.hcclType.c_str());
        return HCCL_E_INTERNAL;
    }

    if (opInfo.hcclType == HCCL_TYPE_ALLREDUCE) {
        CHK_RET(EnqueueAllreduce(opInfo, callback));
    } else if (opInfo.hcclType == HCCL_TYPE_BROADCAST) {
        CHK_RET(EnqueueBroadcast(opInfo, callback));
    } else if (opInfo.hcclType == HCCL_TYPE_ALLGATHER) {
        CHK_RET(EnqueueAllGather(opInfo, callback));
    } else if (opInfo.hcclType == HCCL_TYPE_REDUCESCATTER) {
        CHK_RET(EnqueueReduceScatter(opInfo, callback));
    } else {
        HCCL_ERROR("[HcomExec][EnqueueOperation]Unsupported operation type, hcclType=%s", opInfo.hcclType.c_str());
        return HCCL_E_NOT_SUPPORT;
    }

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueRemoteOperation(HcomRemoteOperation_t opInfo, StatusCallback callback)
{
    // 对opInfo判空
    CHK_PTR_NULL(opInfo.opdesc);
    CHK_PTR_NULL(opInfo.keyAddr);
    CHK_PTR_NULL(opInfo.tableId);
    CHK_PTR_NULL(opInfo.value);

    ge::OpDesc *opdesc = (ge::OpDesc *)opInfo.opdesc;
    std::string hcclType = opdesc->GetType();
    if (parralMap_[MsgQueueType::OPBASE_QUEUE].shutDown) {
        HCCL_ERROR("[HcomExec][EnqueueRemoteOperation]thread has shut down, Enqueue remote hcom operation fail, "\
            "op type[%s]", hcclType.c_str());
        return HCCL_E_INTERNAL;
    }

    if (hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_LOOKUP || hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_UPDATE ||
        hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_LOOKUP_PAIRED ||
        hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_LOOKUP_UNIQUED_PAIRED ||
        hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_UPDATE_PAIRED) {
        CHK_RET(EnqueueRemoteLookupUpdate(opInfo, callback));
    } else {
        HCCL_ERROR("[HcomExec][EnqueueRemoteOperation]Unsupported operation type, hcclType=%s", hcclType.c_str());
        return HCCL_E_NOT_SUPPORT;
    }
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueRemoteAccess(const std::string& remoteAccessType,
    const std::vector<HcomRemoteAccessAddrInfo>& addrInfos, StatusCallback callback)
{
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream);
    if (parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].shutDown) {
        HCCL_ERROR("[Exec][EnqueueRemoteAccess]Thread has been shut down, Enqueue Remote Access failed");
        return HCCL_E_INTERNAL;
    }
    CHK_PRT_RET(addrInfos.empty(),
        HCCL_ERROR("[Exec][EnqueueRemoteAccess]errNo[0x%016llx] in HcomExecEnqueueRemoteAccess addrInfos is empty",\
            HCCL_ERROR_CODE(HCCL_E_PARA)),
        HCCL_E_PARA);

    if (remoteAccess_ == nullptr) {
        if (deviceType_ != DevType::DEV_TYPE_910B && deviceType_ != DevType::DEV_TYPE_910_93) { // 910_93场景临时使用SDMA模拟RDMA
            CHK_RET(HcomGetRemoteAccess(remoteAccess_));
        }
    }

    if (remoteAccessType.compare(HCOM_REMOTE_READ) != 0 &&
        remoteAccessType.compare(HCOM_REMOTE_WRITE) != 0 &&
        remoteAccessType.compare(HCOM_REMOTE_REF_READ) != 0 &&
        remoteAccessType.compare(HCOM_REMOTE_SCATTER_WRITE) != 0) {
        HCCL_ERROR("[Exec][EnqueueRemoteAccess]Invalid Remote Access Type [%s]", remoteAccessType.c_str());
        return HCCL_E_PARA;
    }
    ExecutorMessage message;
    message.SetRemoteAccessMessage(remoteAccessType);
    message.SetStatusCallback(callback);

    // 为了减少拷贝耗时addrInfos不进入message
    // T需要校验 remoteAddr 的有效性，如果是0xFFFF FFFF FFFF FFFF 则不入队
    size_t infoSize = addrInfos.size();
    size_t index;
    const u64 invalidAddr = 0xFFFFFFFFFFFFFFFF;  // 0xFFFFFFFFFFFFFFFF 无效地址常量
    for (index = infoSize - 1; index != 0; index--) {
        if (addrInfos[index].remoteAddr != invalidAddr) {
            break;
        }
    }
    std::vector<HcomRemoteAccessAddrInfo> vct(addrInfos.begin(), addrInfos.begin() + index + 1);
    // 全局变量加锁
    std::unique_lock<std::mutex> lock(msgMutex_);
    listRemoteAccessInfos.push_back(std::move(vct));
    parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].messageQueue.push(std::move(message));
    lock.unlock();

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteAlltoAll(const HcomAllToAllVParams& opInfo, bool isGatherAlltoAll)
{
    HcclWorkflowMode lastWorkflowMode = GetWorkflowMode();
    CHK_RET(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));

    if (!isGatherAlltoAll) {
        u32 rankSize = 0;
        CHK_RET(HcomGetRankSize(opInfo.group, &rankSize));
        HostMem sendCounts = HostMem::alloc(rankSize * sizeof(u64));
        CHK_RET(hrtMemSyncCopy(sendCounts.ptr(), rankSize * sizeof(u64), opInfo.sendcounts, rankSize * sizeof(u64),
            HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
        HostMem sdispls = HostMem::alloc(rankSize * sizeof(u64));
        CHK_RET(hrtMemSyncCopy(sdispls.ptr(), rankSize * sizeof(u64), opInfo.sdispls, rankSize * sizeof(u64),
            HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
        HostMem recvCounts = HostMem::alloc(rankSize * sizeof(u64));
        CHK_RET(hrtMemSyncCopy(recvCounts.ptr(), rankSize * sizeof(u64), opInfo.recvcounts, rankSize * sizeof(u64),
            HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
        HostMem rdispls = HostMem::alloc(rankSize * sizeof(u64));
        CHK_RET(hrtMemSyncCopy(rdispls.ptr(), rankSize * sizeof(u64), opInfo.rdispls, rankSize * sizeof(u64),
            HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
        CHK_RET(HcclAlltoAllV(opInfo.sendbuf, sendCounts.ptr(), sdispls.ptr(), opInfo.sendtype,
                              opInfo.recvbuf, recvCounts.ptr(), rdispls.ptr(), opInfo.recvtype,
                              comm, parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));
    } else {
        CHK_RET(HcclAlltoAllV(opInfo.sendbuf, opInfo.sendcounts, opInfo.sdispls, opInfo.sendtype,
                              opInfo.recvbuf, opInfo.recvcounts, opInfo.rdispls, opInfo.recvtype,
                              comm, parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));
    }
    CHK_RET(SetWorkflowMode(lastWorkflowMode));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteAlltoAllVC(const HcomAllToAllVCParams& opInfo)
{
    HcclWorkflowMode lastWorkflowMode = GetWorkflowMode();
    CHK_RET(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    u32 rankSize = 0;
    CHK_RET(HcomGetRankSize(opInfo.group, &rankSize));
    HostMem sendCountMatrix = HostMem::alloc(rankSize * rankSize * sizeof(u64));
    CHK_RET(hrtMemSyncCopy(sendCountMatrix.ptr(), rankSize * rankSize * sizeof(u64),
        opInfo.sendcountmatrix, rankSize * rankSize * sizeof(u64),
        HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
    CHK_RET(HcclAlltoAllVC(opInfo.sendbuf, sendCountMatrix.ptr(), opInfo.sendtype,
                           opInfo.recvbuf, opInfo.recvtype,
                           comm, parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));

    CHK_RET(SetWorkflowMode(lastWorkflowMode));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteOperation(const HcomOperation_t &opInfo)
{
    HcclResult ret = HCCL_SUCCESS;
    HcclWorkflowMode lastWorkflowMode = GetWorkflowMode();
    if (lastWorkflowMode >= HcclWorkflowMode::HCCL_WORKFLOW_MODE_RESERVED ||
        lastWorkflowMode < HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
        HCCL_ERROR("[Execute][Operation]WorkflowMode[%d] invalid", lastWorkflowMode);
        return HCCL_E_INTERNAL;
    }
    CHK_RET(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
    if (opInfo.hcclType == HCCL_TYPE_ALLREDUCE) {
        ret = ExecuteAllreduce(opInfo);
    } else if (opInfo.hcclType == HCCL_TYPE_BROADCAST) {
        ret = ExecuteBroadcast(opInfo);
    } else if (opInfo.hcclType == HCCL_TYPE_ALLGATHER) {
        ret = ExecuteAllGather(opInfo);
    } else if (opInfo.hcclType == HCCL_TYPE_REDUCESCATTER) {
        ret = ExecuteReduceScatter(opInfo);
    } else {
        HCCL_ERROR("[Execute][Operation]Unsupported operation type, op_type=%s", opInfo.hcclType.c_str());
        ret = HCCL_E_NOT_SUPPORT;
    }
    CHK_RET(SetWorkflowMode(lastWorkflowMode));
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[Execute][Operation]execute operation fail, op_type=%s", opInfo.hcclType.c_str());
    }
    return ret;
}

HcclResult HcomExecutor::ExecuteRemoteOperation(const HcomRemoteOperationParams &opInfo)
{
    // opInfo判空
    std::string hcclType = opInfo.hcclType;
    HcclResult ret = HCCL_SUCCESS;

    HcclWorkflowMode lastWorkflowMode = GetWorkflowMode();
    if (lastWorkflowMode >= HcclWorkflowMode::HCCL_WORKFLOW_MODE_RESERVED ||
        lastWorkflowMode < HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
        HCCL_ERROR("[Execute][RemoteOperation]WorkflowMode[%d] invalid", lastWorkflowMode);
        return HCCL_E_INTERNAL;
    }

    CHK_RET(SetWorkflowMode(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));

    if (hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_LOOKUP_PAIRED) {
        ret = HcclCollRemoteLookupOrUpdatePairedOrUniqued(opInfo, EsOpType::LOOKUP);
    } else if (hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_LOOKUP_UNIQUED_PAIRED) {
        ret = HcclCollRemoteLookupOrUpdatePairedOrUniqued(opInfo, EsOpType::LOOKUP_UNIQUED);
    } else if (hcclType == HCCL_KERNEL_OP_TYPE_COLL_REMOTE_UPDATE_PAIRED) {
        ret = HcclCollRemoteLookupOrUpdatePairedOrUniqued(opInfo, EsOpType::UPDATE);
    } else {
        HCCL_ERROR("[Execute][RemoteOperation]Unsupported remote operation type, op_type=%s", hcclType.c_str());
        ret = HCCL_E_NOT_SUPPORT;
    }

    CHK_RET(SetWorkflowMode(lastWorkflowMode));
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[Execute][RemoteOperation]execute remote operation fail, op_type=%s", hcclType.c_str());
    }
    return ret;
}

HcclResult HcomExecutor::ExecuteBroadcast(const HcomOperation_t &opInfo)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(HcclBroadcast(opInfo.inputPtr, opInfo.count, opInfo.dataType, opInfo.root, comm,
        parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteAllreduce(const HcomOperation_t &opInfo)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(HcclAllReduce(opInfo.inputPtr, opInfo.outputPtr, opInfo.count, opInfo.dataType, opInfo.opType,
        comm, parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteAllGather(const HcomOperation_t &opInfo)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(HcclAllGather(opInfo.inputPtr, opInfo.outputPtr, opInfo.count, opInfo.dataType, comm,
        parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteReduceScatter(const HcomOperation_t &opInfo)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(HcclReduceScatter(opInfo.inputPtr, opInfo.outputPtr, opInfo.count, opInfo.dataType, opInfo.opType,
        comm, parralMap_[MsgQueueType::OPBASE_QUEUE].stream->ptr()));

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::ExecuteRemoteAccess(const RemoteAccessInfo& remoteAccessInfo)
{
    CHK_PRT_RET(listRemoteAccessInfos.empty(),
        HCCL_ERROR("[Execute][RemoteAccess]errNo[0x%016llx] listRemoteAccessInfos is empty",
        HCOM_ERROR_CODE(HCCL_E_PARA)), HCCL_E_PARA);

    if (remoteAccessInfo.remoteAccessType.compare(HCOM_REMOTE_READ) == 0 ||
        remoteAccessInfo.remoteAccessType.compare(HCOM_REMOTE_REF_READ) == 0) {
        if (deviceType_ != DevType::DEV_TYPE_910B && deviceType_ != DevType::DEV_TYPE_910_93) { // 910_93场景临时使用SDMA模拟RDMA
            CHK_RET(remoteAccess_->RemoteRead(listRemoteAccessInfos.front(), \
                parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream->ptr()));
        } else {
            for (u32 i = 0; i < listRemoteAccessInfos.front().size(); i++) {
                CHK_RET(hrtMemAsyncCopy(
                    reinterpret_cast<void *>(static_cast<uintptr_t>(listRemoteAccessInfos.front()[i].localAddr)),
                    listRemoteAccessInfos.front()[i].length,
                    reinterpret_cast<void *>(static_cast<uintptr_t>(listRemoteAccessInfos.front()[i].remoteAddr)),
                    listRemoteAccessInfos.front()[i].length, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE,
                    parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream->ptr()));
            }
        }
    } else {
        if (deviceType_ != DevType::DEV_TYPE_910B && deviceType_ != DevType::DEV_TYPE_910_93) { // 910_93场景临时使用SDMA模拟RDMA
            CHK_RET(remoteAccess_->RemoteWrite(listRemoteAccessInfos.front(), \
                parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream->ptr()));
        } else {
            for (u32 i = 0; i < listRemoteAccessInfos.front().size(); i++) {
                CHK_RET(hrtMemAsyncCopy(
                    reinterpret_cast<void *>(static_cast<uintptr_t>(listRemoteAccessInfos.front()[i].remoteAddr)),
                    listRemoteAccessInfos.front()[i].length,
                    reinterpret_cast<void *>(static_cast<uintptr_t>(listRemoteAccessInfos.front()[i].localAddr)),
                    listRemoteAccessInfos.front()[i].length, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST,
                    parralMap_[MsgQueueType::REMOTE_ACCESS_QUEUE].stream->ptr()));
            }
        }
    }

    std::unique_lock<std::mutex> lock(msgMutex_);
    listRemoteAccessInfos.pop_front();
    lock.unlock();
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::MessageProcessThreadLoop(MsgQueueType queType)
{
    // 给当前线程添加名字
    SetThreadName("Hccl_MessageProcess");

    HCCL_INFO("MessageProcessThreadLoop start queType:[%d]", queType);
    HcclResult rtRet = hrtSetDevice(deviceLogicId_);
    if (rtRet != HCCL_SUCCESS) {
        HCCL_ERROR("[Process][ThreadLoop]errNo[0x%016llx] set device error", HCOM_ERROR_CODE(rtRet));
        parralMap_[queType].shutDown = true;
        return rtRet;
    }
    SetWorkflowMode(workflowMode_);

    HcclResult ret = HCCL_SUCCESS;

    while (!parralMap_[queType].shutDown) {
        if (parralMap_[queType].messageQueue.empty()) {
            std::this_thread::sleep_for(std::chrono::microseconds(THREAD_SLEEP_DURATION_US));
            continue;
        }

        ExecutorMessage message = PopMessagesFromQueue(queType);
        StatusCallback statusCallback = message.GetStatusCallback();
        if (message.GetMessageType() == ExecutorMessageType::MSG_OPBASED) {
            ret = ExecuteOperation(message.GetOperationMessage());
            HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]execute operation error, ret[%d]", ret),
                statusCallback, parralMap_[queType].shutDown = true);
        } else if (message.GetMessageType() == ExecutorMessageType::MSG_REMOTE_ACCESS) {
            ret = ExecuteRemoteAccess(message.GetRemoteAccessMessage());
            HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]execute remote access error, ret[%d]", ret),
                statusCallback, parralMap_[queType].shutDown = true);
        } else if (message.GetMessageType() == ExecutorMessageType::MSG_ALLTOALL) {
            ret = ExecuteAlltoAll(message.GetAlltoAllMessage(), message.isGatherAlltoAll_);
            if (message.isGatherAlltoAll_) {
                HcomAllToAllVParams opInfo = message.GetAlltoAllMessage();
                DeleteGatherAlltoAllMem(opInfo);
            }
            HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]execute all to all error, ret[%d]", ret),
                statusCallback, parralMap_[queType].shutDown = true);
        } else if (message.GetMessageType() == ExecutorMessageType::MSG_ALLTOALLVC) {
            ret = ExecuteAlltoAllVC(message.GetAlltoAllVCMessage());
            HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]execute all to all vc error, ret[%d]", ret),
                statusCallback, parralMap_[queType].shutDown = true);
        } else if (message.GetMessageType() == ExecutorMessageType::MSG_REMOTE_OP) {
            ret = ExecuteRemoteOperation(message.GetRemoteOperationMessage());
            HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]execute remote operation error, ret[%d]", \
                ret),
                statusCallback, parralMap_[queType].shutDown = true);
        }

        // 每个Hcom算子需申请一个event同步
        ret = EnqueueEvent(parralMap_[queType].stream->ptr(), statusCallback, queType);
        HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]enqueue event error, ret[%d]", ret),
            statusCallback, parralMap_[queType].shutDown = true);

        ret = hcclStreamSynchronize(parralMap_[queType].stream->ptr());
        HCOM_EXECUTOR_ERR_BREAK(ret, HCCL_ERROR("[Process][ThreadLoop]hcclStreamSynchronize error, ret[%d]", ret),
            statusCallback, parralMap_[queType].shutDown = true);
    }

    CHK_RET(hrtResetDevice(deviceLogicId_));

    return ret;
}

HcclResult HcomExecutor::CallbackNotifyThreadLoop(MsgQueueType queType)
{
    // 给当前线程添加名字
    SetThreadName("Hccl_Notify");

    HcclResult rtRet = hrtSetDevice(deviceLogicId_);
    if (rtRet != HCCL_SUCCESS) {
        HCCL_ERROR("[Callback][NotifyThreadLoop]errNo[0x%016llx] set device error", HCOM_ERROR_CODE(rtRet));
        parralMap_[queType].shutDown = true;
        return rtRet;
    }
    SetWorkflowMode(workflowMode_);

    HcclResult status = HCCL_SUCCESS;
    while (!parralMap_[queType].shutDown) {
        if (parralMap_[queType].eventQueue.empty()) {
            std::this_thread::sleep_for(std::chrono::microseconds(THREAD_SLEEP_DURATION_US));
            continue;
        }
        EventInfo_t eventInfo = PopEventFromQueue(queType);
        auto startTime = std::chrono::steady_clock::now();
        auto timeout = std::chrono::seconds(EVENT_QUERY_TIMEOUT_S);
        bool bTimeout = false;
        HcclResult ret;
        while (true) {
            ret = hrtEventQuery(eventInfo.event);
            if (ret == HCCL_SUCCESS) {
                break;
            }
            bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
            if (bTimeout) {
                HCCL_ERROR("[Callback][NotifyThreadLoop]Event query timeout");
                break;
            }
            std::this_thread::sleep_for(std::chrono::microseconds(THREAD_SLEEP_DURATION_US));
        }

        ret = hrtEventDestroy(eventInfo.event);
        status = (ret != HCCL_SUCCESS || bTimeout) ? HCCL_E_RUNTIME : HCCL_SUCCESS;
        HCOM_EXECUTOR_ERR_BREAK(status,
            HCCL_ERROR("[Callback][NotifyThreadLoop]event execute failed, event destroy ret[%d], is timeout[%d]",
                ret, bTimeout), eventInfo.callback, parralMap_[queType].shutDown = true);
        // 执行成功，回调返回 SUCCESS
        eventInfo.callback(HCCL_SUCCESS);
        HCCL_RUN_INFO("[Callback][NotifyThreadLoop]hcom executor callback HCCL_SUCCESS, message type[%d]", queType);
    }

    CHK_RET(hrtResetDevice(deviceLogicId_));

    return status;
}

HcclResult HcomExecutor::EnqueueRemoteLookupUpdate(HcomRemoteOperation_t opInfo, StatusCallback callback)
{
    CHK_PTR_NULL(callback);

    ExecutorMessage message;

    HcomRemoteOperationParams opParams{};
    opParams.keyAddr = opInfo.keyAddr;
    opParams.tableId = opInfo.tableId;

    opParams.keyNumInput = opInfo.keyNumInput;
    opParams.uniqueIndices = opInfo.uniqueIndices;
    opParams.keyCount = opInfo.keyCount;

    opParams.value = opInfo.value;
    opParams.indices = opInfo.indices;
    opParams.numUniqued = opInfo.numUniqued;
    opParams.psSeg = opInfo.psSeg;
    opParams.psSegNum = opInfo.psSegNum;
    ge::OpDesc *opdesc = (ge::OpDesc *)opInfo.opdesc;
    CHK_RET(HcclGetRemoteOperationParams(*opdesc, &opParams));

    message.SetRemoteOperationMessage(opParams);
    message.SetStatusCallback(callback);
    PushMessageToQueue(message, MsgQueueType::OPBASE_QUEUE);

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::EnqueueBroadcast(HcomOperation_t opInfo, StatusCallback callback)
{
    // op信息合法性校验
    CHK_PTR_NULL(callback);
    CHK_PTR_NULL(opInfo.inputPtr);

    CHK_RET(HcomCheckCount(opInfo.count));
    CHK_RET(HcomCheckDataType(opInfo.dataType));

    ExecutorMessage message;
    message.SetOperationMessage(opInfo);
    message.SetStatusCallback(callback);
    PushMessageToQueue(message, MsgQueueType::OPBASE_QUEUE);

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::EnqueueAllreduce(HcomOperation_t opInfo, StatusCallback callback)
{
    // op信息合法性校验
    CHK_PTR_NULL(callback);
    CHK_PTR_NULL(opInfo.inputPtr);
    CHK_PTR_NULL(opInfo.outputPtr);

    CHK_RET(HcomCheckCount(opInfo.count));
    CHK_RET(HcomCheckDataType(opInfo.dataType));
    CHK_RET(HcomCheckReductionOp(opInfo.opType));

    ExecutorMessage message;
    message.SetOperationMessage(opInfo);
    message.SetStatusCallback(callback);
    PushMessageToQueue(message, MsgQueueType::OPBASE_QUEUE);

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::EnqueueReduceScatter(HcomOperation_t opInfo, StatusCallback callback)
{
    // op信息合法性校验
    CHK_PTR_NULL(callback);
    CHK_PTR_NULL(opInfo.inputPtr);
    CHK_PTR_NULL(opInfo.outputPtr);

    CHK_RET(HcomCheckCount(opInfo.count));
    CHK_RET(HcomCheckDataType(opInfo.dataType));
    CHK_RET(HcomCheckReductionOp(opInfo.opType));

    ExecutorMessage message;
    message.SetOperationMessage(opInfo);
    message.SetStatusCallback(callback);
    PushMessageToQueue(message, MsgQueueType::OPBASE_QUEUE);

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::EnqueueAllGather(HcomOperation_t opInfo, StatusCallback callback)
{
    // op信息合法性校验
    CHK_PTR_NULL(callback);
    CHK_PTR_NULL(opInfo.inputPtr);
    CHK_PTR_NULL(opInfo.outputPtr);

    CHK_RET(HcomCheckCount(opInfo.count));
    CHK_RET(HcomCheckDataType(opInfo.dataType));

    ExecutorMessage message;
    message.SetOperationMessage(opInfo);
    message.SetStatusCallback(callback);
    PushMessageToQueue(message, MsgQueueType::OPBASE_QUEUE);

    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::EnqueueEvent(HcclRtStream stream, StatusCallback callback, MsgQueueType queType)
{
    HcclResult ret;
    HcclRtEvent event = nullptr;
    CHK_RET(hrtEventCreate(&event));

    ret = hrtEventRecord(event, stream);
    if (ret != HCCL_SUCCESS) {
        HCCL_ERROR("[Enqueue][Event]rt_event_record failed, ret[%d]", ret);
        ret = hrtEventDestroy(event);
        event = nullptr;
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Enqueue][Event]rt_event_destroy failed, ret[%d]", ret), ret);
    }

    auto startTime = std::chrono::steady_clock::now();
    auto timeout = std::chrono::seconds(EVENT_QUERY_TIMEOUT_S);
    bool bTimeout = false;
    // 此处查询不能加锁，避免死锁
    while (parralMap_[queType].eventQueue.size() >= MAX_ENQUEUE_SIZE) {
        bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
        CHK_PRT_RET(bTimeout, HCCL_ERROR("[Enqueue][Event]Event enqueue timeout, size[%d]",
                                         parralMap_[queType].eventQueue.size()), HCCL_E_TIMEOUT);
        std::this_thread::sleep_for(std::chrono::microseconds(THREAD_SLEEP_DURATION_US));
    }

    // event以及回调信息入栈, 多线程加锁
    std::lock_guard<std::mutex> guard(eventMutex_);
    parralMap_[queType].eventQueue.push({ event, callback });

    return HCCL_SUCCESS;
}

ExecutorMessage HcomExecutor::PopMessagesFromQueue(MsgQueueType queueType)
{
    std::lock_guard<std::mutex> guard(msgMutex_);
    ExecutorMessage message = parralMap_[queueType].messageQueue.front();
    parralMap_[queueType].messageQueue.pop();

    return message;
}

EventInfo_t HcomExecutor::PopEventFromQueue(MsgQueueType queueType)
{
    std::lock_guard<std::mutex> guard(eventMutex_); // 多线程访问加锁
    EventInfo_t eventInfo = parralMap_[queueType].eventQueue.front();
    parralMap_[queueType].eventQueue.pop();

    return eventInfo;
}

void HcomExecutor::PushMessageToQueue(ExecutorMessage &message, MsgQueueType queueType)
{
    std::unique_lock<std::mutex> lock(msgMutex_);
    parralMap_[queueType].messageQueue.push(std::move(message));
    lock.unlock();
}

HcclResult HcomExecutor::GetComm(const char *group, HcclComm *comm)
{
    CHK_PTR_NULL(comm);
    std::shared_ptr<hcclComm> hcclComm;
    CHK_RET(HcomGetCommByGroup(group, hcclComm));
    *comm = static_cast<HcclComm>(hcclComm.get());
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueAllToAllV(HcomAllToAllVParams opInfo, StatusCallback callback)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(InitHcclComm(opInfo.group, comm));
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::OPBASE_QUEUE].stream);
    if (parralMap_[MsgQueueType::OPBASE_QUEUE].shutDown) {
        HCCL_ERROR("[Exec][EnqueueAlltoAll]Thread has been shut down, Enqueue all to all failed");
        return HCCL_E_INTERNAL;
    }

    CHK_PTR_NULL(opInfo.sendcounts);
    CHK_PTR_NULL(opInfo.sdispls);
    CHK_PTR_NULL(opInfo.rdispls);
    CHK_PTR_NULL(opInfo.recvcounts);

    ExecutorMessage message;
    message.SetAlltoAllMessage(opInfo);
    message.SetStatusCallback(callback);
    std::unique_lock<std::mutex> lock(msgMutex_);
    parralMap_[MsgQueueType::OPBASE_QUEUE].messageQueue.push(std::move(message));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueAllToAllVC(HcomAllToAllVCParams opInfo, StatusCallback callback)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(opInfo.group, &comm));
    CHK_RET(InitHcclComm(opInfo.group, comm));
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::OPBASE_QUEUE].stream);
    if (parralMap_[MsgQueueType::OPBASE_QUEUE].shutDown) {
        HCCL_ERROR("[Exec][EnqueueAlltoAllVC]Thread has been shut down, Enqueue all to all vc failed");
        return HCCL_E_INTERNAL;
    }

    CHK_PTR_NULL(opInfo.sendcountmatrix);

    ExecutorMessage message;
    message.SetAlltoAllVCMessage(opInfo);
    message.SetStatusCallback(callback);
    std::unique_lock<std::mutex> lock(msgMutex_);
    parralMap_[MsgQueueType::OPBASE_QUEUE].messageQueue.push(std::move(message));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::HcomExecEnqueueGatherAllToAllV(HcomGatherAllToAllVParams params,
    std::function<void(HcclResult status)> callback)
{
    // 获取通信域
    HcclComm comm;
    CHK_RET(GetComm(params.group, &comm));
    CHK_RET(InitHcclComm(params.group, comm));
    CHK_SMART_PTR_NULL(parralMap_[MsgQueueType::OPBASE_QUEUE].stream);
    if (parralMap_[MsgQueueType::OPBASE_QUEUE].shutDown) {
        HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV]Thread has been shut down, Enqueue all to all failed");
        return HCCL_E_INTERNAL;
    }

    CHK_PTR_NULL(params.addrInfoCountPerRank);
    CHK_PTR_NULL(params.recvcounts);
    CHK_PTR_NULL(params.gatheredbuf);
    CHK_PTR_NULL(params.rdispls);

    std::unique_lock<std::mutex> lock(msgMutex_);
    u32 rankSize = 0;
    CHK_RET(HcomGetRankSize(params.group, &rankSize));
    std::vector<u64> addrInfoCountPerRank(rankSize, 0);
    CHK_RET(hrtMemSyncCopy(addrInfoCountPerRank.data(), rankSize * sizeof(u64), params.addrInfoCountPerRank,
        rankSize * sizeof(u64), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));

    u64 blockNum = 0;
    for (u32 index = 0; index < rankSize; index++) {
        blockNum += addrInfoCountPerRank[index];
    }
    if (blockNum != 0) {
        CHK_PTR_NULL(params.addrInfo);
    }
    std::vector<u64> addrInfo(blockNum * NUM_TWO, 0);
    CHK_RET(hrtMemSyncCopy(addrInfo.data(), addrInfo.size() * sizeof(u64), params.addrInfo,
        addrInfo.size() * sizeof(u64), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));

    // 执行gather
    u64 *sendCounts;
    u64 *sdispls;
    void *recvCounts;
    void *rdispls;
    CHK_RET(AllocGatherAlltoAllMem(sendCounts, sdispls, recvCounts, rdispls, rankSize));

    PushGatherAlltoAllParaToQue(params, sendCounts, sdispls, recvCounts, rdispls, callback);

    CHK_RET(RunGather(addrInfo, addrInfoCountPerRank, rankSize, sendCounts, sdispls, params.gatheredbuf,
        params.addrLength));

    auto ret = memset_s(recvCounts, rankSize * sizeof(u64), 0, rankSize * sizeof(u64));
    CHK_PRT_RET(ret != EOK, HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV] mem set failed, count[%lld]",
        rankSize * sizeof(u64)), HCCL_E_SYSCALL);
    CHK_RET(hrtMemSyncCopy(recvCounts, rankSize * sizeof(u64), params.recvcounts, rankSize * sizeof(u64),
                           HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));

    ret = memset_s(rdispls, rankSize * sizeof(u64), 0, rankSize * sizeof(u64));
    CHK_PRT_RET(ret != EOK, HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV] mem set failed, count[%lld]",
        rankSize * sizeof(u64)), HCCL_E_SYSCALL);
    CHK_RET(hrtMemSyncCopy(rdispls, rankSize * sizeof(u64), params.rdispls, rankSize * sizeof(u64),
                           HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
    return HCCL_SUCCESS;
}

HcclResult HcomExecutor::RunGather(std::vector<u64> &addrInfo, std::vector<u64> &addrInfoCountPerRank, u32 rankSize,
                                   u64 *sendCounts, u64 *sdispls, void *sendDevBuf, s32 addrLength)
{
    u64 memSize = 0;
    u64 perThreadCount = addrInfo.size() / NUM_TWO / GATHER_THREAD_NUM;
    std::vector<u64> perThreadCounts(GATHER_THREAD_NUM, perThreadCount);
    perThreadCounts[GATHER_THREAD_NUM - 1] = addrInfo.size() / NUM_TWO - perThreadCount * (GATHER_THREAD_NUM - 1);
    std::vector<u64> offset(GATHER_THREAD_NUM, 0);
    if (addrLength == -1) { // 数据包长度不一样的情况
        u32 offsetIndex = 0;
        for (u32 index = 1; index < addrInfo.size(); index += NUM_TWO) { // 由于是二元组，单数为数据包的长度，每个循环+2
            /* 如果数据包数量小于线程数量则offset全置为0 */
            if (perThreadCount != 0) {
                /* 条件1：当累加的数量达到perThreadCount时往offset中填入累加值，即可计算出前面thread产生的offset值 */
                /* 条件2：由于第0个thread的offset为0，后面的线程的offset为前面线程处理数据量的累加，因此对最后一个值弃之不用 */
                if (index / NUM_TWO % perThreadCount == 0 && offsetIndex < GATHER_THREAD_NUM) {
                    offset[offsetIndex] = memSize;
                    offsetIndex++;
                }
            }
            memSize += addrInfo[index];
        }
    } else {
        memSize = addrInfo.size() / NUM_TWO * addrInfo[1];
        for (u32 index = 0; index < GATHER_THREAD_NUM; index++) {
            offset[index] = index * perThreadCount * addrInfo[1];
        }
    }

    // 多线程拷贝
    HostMem tmpHostMem = HostMem::alloc(memSize);
    std::vector<std::unique_ptr<std::thread>> threads(GATHER_THREAD_NUM);
    for (u32 num = 0; num < GATHER_THREAD_NUM; num++) {
        threads[num].reset(new (std::nothrow) std::thread(&HcomExecutor::GatherMemCopyThread, this, tmpHostMem.ptr(),
            offset[num], std::ref(addrInfo), num * perThreadCount * NUM_TWO, perThreadCounts[num], memSize));
        CHK_PRT_RET(!threads[num], HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV]threads[%u] reset "\
            "failed ", num), HCCL_E_INTERNAL);
    }

    // 构造入参
    auto ret = memset_s(sendCounts, rankSize * sizeof(u64), 0, rankSize * sizeof(u64));
    CHK_PRT_RET(ret != EOK, HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV] mem set failed,count[%lld]",
        rankSize * sizeof(u64)), HCCL_E_SYSCALL);
    u64 prevNum = 0;
    u64 nextNum = 0;
    for (u32 index = 0; index < addrInfoCountPerRank.size(); index++) {
        nextNum += addrInfoCountPerRank[index];
        for (u64 i = NUM_TWO * prevNum; i < NUM_TWO * nextNum; i += NUM_TWO) {
            *(sendCounts + index) += addrInfo[i + 1];
        }
        prevNum = nextNum;
    }

    ret = memset_s(sdispls, rankSize * sizeof(u64), 0, rankSize * sizeof(u64));
    CHK_PRT_RET(ret != EOK, HCCL_ERROR("[Exec][EnqueueGatherAlltoAllV] mem set failed, count[%lld]",
        rankSize * sizeof(u64)), HCCL_E_SYSCALL);
    u64 displ = 0;
    for (u32 i = 0; i < rankSize; i++) {
        *(sdispls + i) = displ;
        displ += *(sendCounts + i);
    }

    // 等待线程执行完毕
    for (u32 num = 0; num < GATHER_THREAD_NUM; num++) {
        threads[num]->join();
    }

    CHK_RET(hrtMemSyncCopy(sendDevBuf, memSize, tmpHostMem.ptr(), memSize,
        HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
    return HCCL_SUCCESS;
}

void HcomExecutor::PushGatherAlltoAllParaToQue(HcomGatherAllToAllVParams &gatherParams, void *sendCounts, void *sdispls,
    void *recvCounts, void *rdispls, std::function<void(HcclResult status)> callback)
{
    HcomAllToAllVParams opInfo;
    opInfo.sendbuf = gatherParams.gatheredbuf;
    opInfo.sendcounts = sendCounts;
    opInfo.sdispls = sdispls;
    opInfo.sendtype = HCCL_DATA_TYPE_INT8;
    opInfo.recvbuf = gatherParams.recvbuf;
    opInfo.recvcounts = recvCounts;
    opInfo.rdispls = rdispls;
    opInfo.recvtype = gatherParams.recvtype;
    opInfo.group = nullptr;

    ExecutorMessage message;
    message.SetAlltoAllMessage(opInfo, true);
    message.SetStatusCallback(callback);
    parralMap_[MsgQueueType::OPBASE_QUEUE].messageQueue.push(std::move(message));
}

void HcomExecutor::GatherMemCopyThread(void *baseAddr, u64 offset, std::vector<u64> &addrInfo, u64 beginIndex,
    u64 count, u64 tmpMemSize)
{
    // 给当前线程添加名字
    SetThreadName("Hccl_GatherMemCopy");

    void *addr = nullptr;
    u64 length = 0;
    auto destMax = [&]()-> u64 {
        return tmpMemSize < offset ? 0 : tmpMemSize - offset;
    };

    for (u32 index = 0; index < count; index++) {
        addr = reinterpret_cast<void *>(addrInfo[beginIndex + NUM_TWO * index]);
        length = addrInfo[beginIndex + index * NUM_TWO + 1];
        if (memcpy_s(static_cast<char *>(baseAddr) + offset, destMax(), addr, length) != EOK) {
            HCCL_ERROR("[MemCopy][GatherAlltoAllV] mem copy failed, destMax[%llu], count[%llu]",
                tmpMemSize - offset, length);
            return;
        }
        offset += length;
    }
}

HcclResult HcomExecutor::AllocGatherAlltoAllMem(u64* &sendCounts, u64* &sdispls, void* &recvCountsPtr,
    void * &rdisplsPtr, u32 rankSize)
{
    HcclResult ret;
    do {
        sendCounts = new (std::nothrow) u64[rankSize];
        CHK_PRT_BREAK(sendCounts == nullptr, HCCL_ERROR("sendCounts new failed"), ret = HCCL_E_PTR);

        sdispls = new (std::nothrow) u64[rankSize];
        CHK_PRT_BREAK(sdispls == nullptr, HCCL_ERROR("sdispls new failed"), ret = HCCL_E_PTR);

        recvCountsPtr = new (std::nothrow) u64[rankSize];
        CHK_PRT_BREAK(recvCountsPtr == nullptr, HCCL_ERROR("recvCountsPtr new failed"), ret = HCCL_E_PTR);

        rdisplsPtr = new (std::nothrow) u64[rankSize];
        CHK_PRT_BREAK(rdisplsPtr == nullptr, HCCL_ERROR("rdisplsPtr new failed"), ret = HCCL_E_PTR);

        HCCL_DEBUG("[AllocGatherAlltoAllMem] new sendCounts, sdispls, recvCounts, rdispls success!");
        return HCCL_SUCCESS;
    } while(0);

    if (sendCounts != nullptr) {
        delete[] sendCounts;
        sendCounts = nullptr;
    }
    if (sdispls != nullptr) {
        delete[] sdispls;
        sdispls = nullptr;
    }
    if (recvCountsPtr != nullptr) {
        delete[] static_cast<u64 *>(recvCountsPtr);
        recvCountsPtr = nullptr;
    }
    if (rdisplsPtr != nullptr) {
        delete[] static_cast<u64 *>(rdisplsPtr);
        rdisplsPtr = nullptr;
    }
    return ret;
}

void HcomExecutor::DeleteGatherAlltoAllMem(HcomAllToAllVParams &params)
{
    if (params.sendcounts != nullptr) {
        delete[] static_cast<u64 *>(params.sendcounts);
        params.sendcounts = nullptr;
    }
    if (params.sdispls != nullptr) {
        delete[] static_cast<u64 *>(params.sdispls);
        params.sdispls = nullptr;
    }
    if (params.recvcounts != nullptr) {
        delete[] static_cast<u64 *>(params.recvcounts);
        params.recvcounts = nullptr;
    }
    if (params.rdispls != nullptr) {
        delete[] static_cast<u64 *>(params.rdispls);
        params.rdispls = nullptr;
    }
}

void HcomExecutor::CleanQueueResources()
{
    if (!parralMap_[MsgQueueType::OPBASE_QUEUE].messageQueue.empty()) {
        for (u32 index = 0; index < parralMap_[MsgQueueType::OPBASE_QUEUE].messageQueue.size(); index++) {
            ExecutorMessage message = PopMessagesFromQueue(MsgQueueType::OPBASE_QUEUE);
            if (message.isGatherAlltoAll_) {
                HcomAllToAllVParams opInfo = message.GetAlltoAllMessage();
                DeleteGatherAlltoAllMem(opInfo);
            }
        }
    }
    parralMap_.clear();
}
}
